/**
 * $Id: SGIPSubmitConnector.java,v 1.33 2006/09/30 05:31:36 martin Exp $
 *
 * Copyright 2006 Skyinfo, Inc. All Rights Reserved.
 */
package com.haoxi.gateway.connector.impl;

import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.EOFException;
import java.io.IOException;
import java.net.Socket;
import java.net.SocketException;


import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.haoxi.gateway.common.CommonMsgBean;

import com.haoxi.gateway.connector.Connector;
import com.haoxi.gateway.runner.RunnerMgr;
import com.haoxi.gateway.sgip12msg.SGIPActiveMessage;
import com.haoxi.gateway.sgip12msg.SGIPActiveRepMessage;
import com.haoxi.gateway.sgip12msg.SGIPBindMessage;
import com.haoxi.gateway.sgip12msg.SGIPBindRepMessage;
import com.haoxi.gateway.sgip12msg.SGIPCommon;
import com.haoxi.gateway.sgip12msg.SGIPMessage;
import com.haoxi.gateway.sgip12msg.SGIPSubmitRepMessage;
import com.haoxi.gateway.sgip12msg.SGIPUnbindMessage;
import com.haoxi.gateway.sgip12msg.SGIPUnbindRepMessage;
import com.haoxi.gateway.sgip12msg.SequenceSower;

/**
 * 实现中国联通下发的连接器
 * 
 * Revision History
 * 
 * 2006-6-17,star.Zhao,created it
 */
public class SGIPSubmitConnector extends Connector {

    private Log log = LogFactory.getLog(SGIPSubmitConnector.class);

    private static final int LOGIN_TYPE = 1;
    private static final int CONNECT_SGIPSMG_SUCCESSFULL = 0;
    private static int errorcount = 0;

    public SGIPSubmitConnector(String aGatewayName, String connName,
            String aICP_ID, String loginName, String aPassword,
            String aGatewayIP, int aGatewayPort, int sVersion,
            String protcolName) {
        super(aGatewayName, connName, aICP_ID, loginName, aPassword,
                aGatewayIP, aGatewayPort, sVersion, protcolName);
    }

    public boolean writeProtcol(CommonMsgBean obj) {
        // 当连接通道被打开并且绑定成功这个时候才能向对方发送消息
        boolean result = false;
        if (isSocketKeepLive && isConnected && (!isReOpen)) {
            try {
                // 这里一定会抛异常
                out.write(obj.getBody());
                out.flush();
                result = true;
            } catch (IOException ex) {
                // 立刻断掉连接进行重新连接
                reOpenConnect();
                result = false;
            }
        } else {
            result = false;
        }
        return result;
    }

    // // 重新连接
    public boolean reOpenConnect() {
        log.debug("begin reOpenConnect()");
        isReOpen = true; // 表示重新连接
        disConnected();
        resetSendMTNoRespCount();
        try {
            Thread.sleep(failConnectSleepTime);
        } catch (InterruptedException e) {
            log.debug("", e);
        }
        openConnect();
        synchronized (this) {
            notifyAll();
        }
        isReOpen = false; // 重新连接完毕
        log.debug("end reOpenConnect()");
        return isConnected;
    }

    // 打开连接
    public synchronized boolean openConnect() {
        in = null;
        out = null;
        soc = null;

        StringBuffer logStrBuff = new StringBuffer();
        logStrBuff.append("gatewayName=").append(gatewayName).append(
                ",protcolName=").append(protcolName).append(",gatewayIP=")
                .append(gatewayIP).append(",port=").append(gatewayPort);
        log.debug(logStrBuff.toString());
        // 向SMC服务器建立连接
        try {
            soc = new Socket(gatewayIP, gatewayPort);
            // 没有抛出异常只能代表握手成功,并不代表连接建立成功
            // 设置超时时间
            soc.setSoTimeout(socketTimeOut);
            // 得到输入输出流
            in = new DataInputStream(new BufferedInputStream(soc
                    .getInputStream()));
            this.out = new DataOutputStream(new BufferedOutputStream(soc
                    .getOutputStream()));
            isSocketKeepLive = true;
            log.debug("socket accept success!!!");
            errorcount = 0;
        } catch (IOException ex) {
            // 有异常，进行重新连接
            isSocketKeepLive = false;
            log.debug("socket accept error!!!", ex);
            try {
                errorcount++;
                // 如果连续尝试30次仍然不能连接成功
                if (errorcount > 30) {
                    // 重起系统
                    System.exit(1);
                }
                Thread.sleep(1000);
            } catch (InterruptedException inter) {
                log.error("sleep error", inter);
            }
            return openConnect();
        }
        // bind请求成功，启动心跳线程
        // checkConnect的返回结果是isConnected
        if (checkConnect(loginName, password)) {
            new SGIPActiveTestThread().start();
        } else {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException inter) {
                log.error("sleep error", inter);
            }
            return openConnect();
        }
        // 绑定成功则开辟这个连接器，否则没必要开启

        return isConnected;
    }

    public CommonMsgBean readProtcol() throws IOException {
        CommonMsgBean result = null;
        int commid = 0;
        int readLength = 0;
        byte[] bodybytes = new byte[0];
        SGIPMessage respMessage = null;
        SGIPMessage reqMessage = null;
        try {
            if (soc != null && isSocketKeepLive && isConnected && soc.isBound()
                    && soc.isConnected() && (!soc.isInputShutdown())
                    && (!soc.isOutputShutdown())) {
                int ch1 = in.read();
                if (ch1 < 0) {
                    return result;
                }
                int ch2 = in.read();
                int ch3 = in.read();
                int ch4 = in.read();
                readLength = ((ch1 << 24) + (ch2 << 16) + (ch3 << 8) + (ch4 << 0));
                commid = in.readInt();
                bodybytes = new byte[readLength - 8];
                in.readFully(bodybytes);
            } else {
                if (!isReOpen) {
                    reOpenConnect();
                }
                return result;
            }
            if (commid == SGIPCommon.SGIP_SUBMIT_RESP) {
                reqMessage = new SGIPSubmitRepMessage(bodybytes);
                log.debug(reqMessage);
                resetSendMTNoRespCount();
                result = new CommonMsgBean();
                String gatewayCode = RunnerMgr.getInstance().getGatewayInfo()
                        .getGatewayCode();
                result.setCommid(SGIPCommon.SGIP_SUBMIT_RESP);
                result.setGateWayCode(gatewayCode);
                result.setConnName(this.connName);
                result.setBody(bodybytes);
            } else if (commid == SGIPCommon.SGIP_ACTIVE_TEST_RESP) {
                reqMessage = new SGIPActiveRepMessage(bodybytes);
                log.debug(reqMessage);
            } else if (commid == SGIPCommon.SGIP_ACTIVE_TEST) {
                reqMessage = new SGIPActiveRepMessage(bodybytes);
                respMessage = new SGIPActiveMessage();
                respMessage.setSequenceId(reqMessage.getSequenceId());
                respMessage.setSrcNodeId(reqMessage.getSrcNodeId());
                respMessage.setTimeStamp(reqMessage.getTimeStamp());
                out.write(respMessage.getBytes());
                out.flush();
            } else if (commid == SGIPCommon.SGIP_BIND_RESP) {
                reqMessage = new SGIPBindRepMessage(bodybytes);
                log.debug(reqMessage);
            } else if (commid == SGIPCommon.SGIP_UNBIND_RESP) {
                reqMessage = new SGIPUnbindRepMessage(bodybytes);
                log.debug(reqMessage);
                if (!isReOpen) {
                    reOpenConnect();
                }
            } else if (commid == SGIPCommon.SGIP_UNBIND) {
                reqMessage = new SGIPUnbindMessage(bodybytes);
                respMessage = new SGIPUnbindRepMessage();
                respMessage.setSequenceId(reqMessage.getSequenceId());
                respMessage.setSrcNodeId(reqMessage.getSrcNodeId());
                respMessage.setTimeStamp(reqMessage.getTimeStamp());
                out.write(respMessage.getBytes());
                out.flush();
                log.debug(respMessage);
                if (!isReOpen) {
                    reOpenConnect();
                }
            }
            // 打印日志消息
            // if (null != reqMessage
            // && (commid != SGIPCommon.SGIP_ACTIVE_TEST_RESP || commid !=
            // SGIPCommon.SGIP_ACTIVE_TEST)) {
            // log.debug(this.connName + "<<<<" + reqMessage.toString());
            // }
            // if (null != respMessage
            // && (commid != SGIPCommon.SGIP_ACTIVE_TEST_RESP || commid !=
            // SGIPCommon.SGIP_ACTIVE_TEST)) {
            // log.debug(this.connName + "<<<<" + respMessage.toString());
            // }

        } catch (IOException ex) {
            if (ex instanceof EOFException) {
                log.info("read the end of the stream has been reached!");
                return null;
            }
            if (ex instanceof SocketException) {
                log.error("catch an SocketException", ex);
            } else {
                log.error("catch an IOException", ex);
            }
            if (!isReOpen) {
                reOpenConnect();
            }
        }
        return result;
    }

    public boolean checkConnect(String loginName, String password) {
        log.debug("checkConnect() start!!" + "isSocketKeepLive=="
                + isSocketKeepLive + "soc.isConnected()=" + soc.isConnected());
        StringBuffer logStrBuff = new StringBuffer();
        byte[] respbytes = new byte[0];
        // 连接被正确建立才开始发bind请求,否则不能发
        if (isSocketKeepLive && soc.isConnected()) {
            logStrBuff.append("gatewayName=").append(gatewayName).append(
                    ",protcolName=").append(protcolName).append(",loginName=")
                    .append(loginName).append(",password=").append(password)
                    .append("ConnName==").append(connName);
            log.debug(logStrBuff.toString());
            SGIPBindMessage bindMessage = new SGIPBindMessage(LOGIN_TYPE,
                    loginName, password);
            // 生成唯一序列号
            SequenceSower.setUnicomSequence(bindMessage);
            try {

                out.write(bindMessage.getBytes());
                out.flush();
                log.debug(bindMessage);
                // 消息长度
                int msgLength = in.readInt();
                // 短连接读取对方响应信息
                // 消息命令字
                int commid = in.readInt();
                respbytes = new byte[msgLength - 8];
                // 读取剩余消息
                in.readFully(respbytes);
                if (SGIPCommon.SGIP_BIND_RESP == commid) {
                    SGIPBindRepMessage bindResp = new SGIPBindRepMessage(
                            respbytes);
                    log.debug(bindResp);
                    // 绑定成功才能发消息
                    if (CONNECT_SGIPSMG_SUCCESSFULL == bindResp.getResult()) {
                        isConnected = true;
                        log.debug(getConnName()
                                + " <<<<<<bind to smc successfuly!");
                    }
                    // 对方将关闭连接,因此先关闭连接
                    else {
                        log.error(getConnName() + "<<<<<<<<bind to smc fail");
                        if (!isReOpen) {
                            reOpenConnect();
                        }
                    }
                }
            } catch (IOException ex) {
                log.error(getConnName() + "<<<<<<<bind to smc exception", ex);
                if (!isReOpen) {
                    reOpenConnect();
                }
            }
        }
        return isConnected;
    }

    /**
     * 链路激活测试
     */
    private class SGIPActiveTestThread extends Thread {
        public void run() {
            activeTest();
        }

        public void activeTest() {
            // 连接并绑定成功才进行下发，否则需要重新启动心跳终止
            SGIPActiveMessage active = null;
            while (isSocketKeepLive && isConnected && (!isReOpen)) {
                try {
                    Thread.sleep(15 * 1000);
                } catch (InterruptedException ite) {
                    continue;
                }
                active = new SGIPActiveMessage();
                SequenceSower.setUnicomSequence(active);
                try {
                    out.write(active.getBytes());
                    out.flush();
                    log.debug(active);
                    // 断开连接
                } catch (IOException ex) {
                    if (!isReOpen) {
                        reOpenConnect();
                    }
                    log.error(connName + "send ActiveTest failure.", ex);
                    break;
                }
            }
        }
    }


}
